package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: guoml
 * @Description: 针对企查查t_ipo_company_main_indicator表和聚源LC_MainIndexNew表进行数据映射开发
 * @Date: 2021/12/7 12:51
 */

public class TIpoCompanyMainIndicator {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_ipo_company_main_indicator").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        //获取中数增量数据表中的新增数据（left join）
        Dataset<Row> insert_data = spark.sql("SELECT md5(concat(100,a.ID)) AS ID,\n" +
                "       b.SecuCode,\n" +
                "       a.BasicEPS,\n" +
                "       a.EPSCut,\n" +
                "       a.DilutedEPS,\n" +
                "       a.NetAssetPS,\n" +
                "       a.AccumulationFundPS,\n" +
                "       a.UndividedProfit,\n" +
                "       a.OperCashFlowPS,\n" +
                "       c.TotalOperatingRevenue AS total_operating_income,\n" +
                "       a.EBIT,\n" +
                "       a.NetProfit,\n" +
                "       a.NetProfitCut,\n" +
                "       a.TORGrowRate,\n" +
                "       a.NPParentCompanyYOY,\n" +
                "       a.NPParentCompanyCutYOY,\n" +
                "       '' AS total_operating_income_growth,\n" +
                "       '' AS net_profit_attributable_rolling_growth,\n" +
                "       '' AS deduction_non_net_profit_growth,\n" +
                "       a.ROEWeighted,\n" +
                "       a.ROE,\n" +
                "       a.ROA,\n" +
                "       a.EBITToTOR,\n" +
                "       a.NPToTOR,\n" +
                "       a.TaxesToTP,\n" +
                "       round(d.AdvanceReceipts/c.OperatingRevenue,4) AS advance_receipts,\n" +
                "       a.SaleServiceCashToOR,\n" +
                "       a.CashRateOfSales,\n" +
                "       a.TotalAssetTRate,\n" +
                "       a.ARTDays,\n" +
                "       a.InventoryTDays,\n" +
                "       a.DebtAssetsRatio,\n" +
                "       a.CurrentLiabilityToTL,\n" +
                "       a.CurrentRatio,\n" +
                "       a.QuickRatio,\n" +
                "       '' AS created_date,\n" +
                "       '' AS updated_date,\n" +
                "       a.EndDate,\n" +
                "       '' AS report_encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" AS isadd\n" +
                "FROM a_dc_ep_ods.LC_MainIndexNew a\n" +
                "INNER JOIN\n" +
                "  (SELECT e.*\n" +
                "   FROM a_dc_ep_ods.SecuMain e\n" +
                "   WHERE e.ListedState = 1\n" +
                "     AND e.ListedSector not in (3,7)\n" +
                "     AND e.SecuCategory = 1) b ON a.CompanyCode = b.CompanyCode\n" +
                "LEFT JOIN\n" +
                "  (SELECT f.*\n" +
                "   FROM\n" +
                "     (SELECT g.*,\n" +
                "             row_number () over (PARTITION BY g.CompanyCode,g.EndDate\n" +
                "                                 ORDER BY g.InfoPublDate DESC,g.IfAdjusted) AS adjust\n" +
                "      FROM a_dc_ep_ods.LC_IncomeStatementAll g\n" +
                "      WHERE g.IfMerged = 1\n" +
                "        AND g.IfComplete = 1\n" +
                "        AND g.IfAdjusted in (1,2)\n" +
                "        AND g.InfoSourceCode IN (110101,\n" +
                "                               110102,\n" +
                "                               110103,\n" +
                "                               110104)) f\n" +
                "   WHERE f.adjust = 1) c ON a.CompanyCode = c.CompanyCode\n" +
                "AND a.EndDate = c.EndDate\n" +
                "LEFT JOIN\n" +
                "  (SELECT a.*\n" +
                "   FROM\n" +
                "     (SELECT a.*,\n" +
                "             row_number () over (PARTITION BY a.CompanyCode,a.EndDate\n" +
                "                                 ORDER BY a.InfoPublDate DESC,a.IfAdjusted) AS adjust\n" +
                "      FROM a_dc_ep_ods.LC_BalanceSheetAll a\n" +
                "      WHERE a.IfMerged = 1\n" +
                "        AND a.IfComplete = 1\n" +
                "        AND a.InfoSourceCode IN (110101,\n" +
                "                               110102,\n" +
                "                               110103,\n" +
                "                               110104)) a\n" +
                "   WHERE a.adjust = 1)d ON a.CompanyCode = d.CompanyCode\n" +
                "AND a.EndDate = d.EndDate").union(spark.sql("SELECT md5(concat(200,a.ID)) as ID,\n" +
                "       b.SecuCode,\n" +
                "       a.BasicEPS,\n" +
                "       f.BasicEPSCut as EPSCut,\n" +
                "       a.DilutedEPS,\n" +
                "       a.NetAssetPS,\n" +
                "       a.AccumulationFundPS,\n" +
                "       a.UndividedProfit,\n" +
                "       a.OperCashFlowPS,\n" +
                "       c.TotalOperateRev as total_operating_income,\n" +
                "       a.EBIT,\n" +
                "       a.NPParentCompanyOwners as NetProfit,\n" +
                "       a.NetProfitCut,\n" +
                "       round((c.TotalOperateRev-e.TotalOperateRev)/e.TotalOperateRev,4) as TORGrowRate,\n" +
                "       a.NPParentCompanyYOY,\n" +
                "       a.NPParentCompanyCutYOY,\n" +
                "       '' as total_operating_income_growth,\n" +
                "       '' as net_profit_attributable_rolling_growth,\n" +
                "       '' as deduction_non_net_profit_growth,\n" +
                "       a.WROE as ROEWeighted,\n" +
                "       a.ROE,\n" +
                "       a.ROA,\n" +
                "       a.EBITToTOR,\n" +
                "       a.NPToTOR,\n" +
                "       a.TaxesToTP,\n" +
                "       round(d.AdvancePayment/c.OperatRevenue,4) as advance_receipts,\n" +
                "       a.SaleServiceCashToOR,\n" +
                "       a.CashRateOfSales,\n" +
                "       a.TotalAssetTRate,\n" +
                "       a.ARTDays,\n" +
                "       a.InventoryTDays,\n" +
                "       a.DebtAssetsRatio,\n" +
                "       a.CurrentLiabilityToTL,\n" +
                "       a.CurrentRatio,\n" +
                "       a.QuickRatio,\n" +
                "       '' as created_date,\n" +
                "       '' as updated_date,\n" +
                "       a.EndDate,\n" +
                "       '' as report_encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" as isadd\n" +
                "FROM\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY Mark) AS rn from a_dc_ep_ods.NQ_MainIndexNew a)a where a.rn = 1)a\n" +
                "INNER JOIN\n" +
                "  (select * from a_dc_ep_ods.NQ_SecuMain where ListedState = 1 and ListedSector =3 and SecuCategory = 1) b on a.CompanyCode = b.CompanyCode\n" +
                "LEFT JOIN\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.NQ_IncStatement a where IfMerged = 1 and IfComplete = 1 and IfAdjusted in (1,2) and InfoSource in (2,3,4,5))a where a.adjust = 1)c on a.CompanyCode = c.CompanyCode and a.EndDate = c.EndDate\n" +
                "LEFT JOIN\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.NQ_BalanceSheet a where IfMerged = 1 and IfComplete = 1 and InfoSource in (2,3,4,5))a where a.adjust = 1)d on a.CompanyCode = d.CompanyCode and a.EndDate = d.EndDate\n" +
                "LEFT JOIN\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDateProfit ORDER BY InfoPublDate desc,IfAdjustProfit) AS adjust from a_dc_ep_ods.NQ_MainAcctData a)a where a.adjust = 1)f on a.CompanyCode = f.CompanyCode and a.EndDate = f.EndDateProfit\n" +
                "LEFT JOIN\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.NQ_IncStatement a where IfMerged = 1 and IfComplete = 1 and IfAdjusted in (1,2) and InfoSource in (2,3,4,5))a where a.adjust = 1)e on c.CompanyCode = e.CompanyCode and date_format(e.EndDate,'yyyy-MM-dd') = add_months(c.EndDate,-12)")).union(spark.sql("SELECT md5(concat(300,a.ID)) as ID,\n" +
                "       b.SecuCode,\n" +
                "       a.BasicEPS,\n" +
                "       e.BasicEPSCut as EPSCut,\n" +
                "       a.DilutedEPS,\n" +
                "       a.NetAssetPS,\n" +
                "       a.AccumulationFundPS,\n" +
                "       a.UndividedProfit,\n" +
                "       a.OperCashFlowPS,\n" +
                "       c.TotalOperatingRevenue as total_operating_income,\n" +
                "       a.EBIT,\n" +
                "       a.NPParentCompanyOwners,\n" +
                "       a.NetProfitCut,\n" +
                "       a.ToOpRevenueGrowRate,\n" +
                "       a.NPParentCompanyYOY,\n" +
                "       a.NPParentCompanyCutYOY,\n" +
                "       '' as total_operating_income_growth,\n" +
                "       '' as net_profit_attributable_rolling_growth,\n" +
                "       '' as deduction_non_net_profit_growth,\n" +
                "       a.ROEWeighted,\n" +
                "       a.ROE,\n" +
                "       a.ROA,\n" +
                "       a.EBITToTOR,\n" +
                "       a.NPToTOR,\n" +
                "       a.TaxesToTP,\n" +
                "       round(d.AdvanceReceipts/c.OperatingRevenue,4) as advance_receipts,\n" +
                "       a.SaleServiceCashToOR,\n" +
                "       a.CashRateOfSales,\n" +
                "       a.TotalAssetTRate,\n" +
                "       a.ARTDays,\n" +
                "       a.InventoryTDays,\n" +
                "       a.DebtAssetsRatio,\n" +
                "       a.CurrentLiabilityToTL,\n" +
                "       a.CurrentRatio,\n" +
                "       a.QuickRatio,\n" +
                "       '' as created_date,\n" +
                "       '' as updated_date,\n" +
                "       a.EndDate,\n" +
                "       '' as report_encrypt_key,\n" +
                "       regexp_replace(current_date(),'-',''),\n" +
                "       \"1\" as isadd\n" +
                "FROM\n" +
                "  (select t.* from (select m.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.LC_STIBMainIndex m where IfMerged = 1 and InfoSourceCode in (110101,110102,110103,110104))t where t.adjust = 1)a\n" +
                "INNER JOIN\n" +
                "  (select * from a_dc_ep_ods.SecuMain where ListedState = 1 and ListedSector =7 and SecuCategory = 1) b on a.CompanyCode = b.CompanyCode\n" +
                "LEFT JOIN\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.LC_STIBIncomeState a where IfMerged = 1 and IfComplete = 1 and IfAdjusted in (1,2) and InfoSourceCode in (110101,110102,110103,110104))a where a.adjust = 1)c on a.CompanyCode = c.CompanyCode and a.EndDate = c.EndDate\n" +
                "LEFT JOIN\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.LC_STIBBalanceSheet a where IfMerged = 1 and IfComplete = 1 and InfoSourceCode in (110101,110102,110103,110104))a where a.adjust = 1)d on a.CompanyCode = d.CompanyCode and a.EndDate = d.EndDate\n" +
                "LEFT JOIN\n" +
                "  (select a.* from (select a.*,row_number ( ) over ( PARTITION BY CompanyCode,EndDate ORDER BY InfoPublDate desc,IfAdjusted) AS adjust from a_dc_ep_ods.LC_STIBMainData a where IfMerged = 1 and IfAdjusted in (1,2) and InfoSourceCode in (110101,110102,110103,110104))a where a.adjust = 1)e on a.CompanyCode = e.CompanyCode and a.EndDate = e.EndDate"));
        insert_data.createOrReplaceTempView("tmp_t_ipo_company_main_indicator1");
        spark.sql("insert overwrite table a_dc_ep_incr.t_ipo_company_main_indicator select * from tmp_t_ipo_company_main_indicator1");
        spark.stop();
    }
}
